package discovery

import (
	"common/config"
	"common/logs"
	"context"
	clientv3 "go.etcd.io/etcd/client/v3"
	"google.golang.org/grpc/attributes"
	"google.golang.org/grpc/resolver"
	"time"
)

type Resolver struct {
	conf        config.EtcdConf  // etcd 配置
	etcdClient  *clientv3.Client // etcd 客户端
	DialTimeout int              // 超时时间
	key         string           // 服务名 etcd://user
	clientConn  resolver.ClientConn
	srvAddrList []resolver.Address // 服务列表
	watchChan   clientv3.WatchChan
	closeChan   chan struct{} // 关闭通道
}

// Build 当 grpc.Dial 的时候，就会同步调用此方法
func (r *Resolver) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
	// 1. 连接 Etcd
	var err error
	r.etcdClient, err = clientv3.New(clientv3.Config{
		Endpoints:   r.conf.Addrs,
		DialTimeout: time.Duration(r.DialTimeout) * time.Second,
	})
	if err != nil {
		logs.Fatal("grpc client connect etcd error:%v", err)
	}
	r.clientConn = cc
	// etcd://user
	r.key = target.URL.Host
	r.closeChan = make(chan struct{}, 1)
	// 2. 同步节点信息
	if err := r.sync(); err != nil {
		return nil, err
	}
	// 3. 监听节点发生变动
	go r.watch()
	return nil, nil
}

func (r *Resolver) Scheme() string {
	return Etcd_Name
}

// sync 在 etcd 中同步节点信息
func (r *Resolver) sync() error {
	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(r.conf.RWTimeout)*time.Second)
	defer cancel()
	// 1. 根据 key 在 etcd 中获取服务
	response, err := r.etcdClient.Get(ctx, r.key, clientv3.WithPrefix())
	if err != nil {
		logs.Error("grpc client get etcd failed,err:%v name:%s", err, r.key)
		return err
	}
	// 2. 构建服务列表
	r.srvAddrList = []resolver.Address{}
	for _, value := range response.Kvs {
		server, err := ParseValue(value.Value)
		if err != nil {
			logs.Error("grpc client parse etcd value failed,err:%v value:%s", err, value.Value)
			continue
		}
		r.srvAddrList = append(r.srvAddrList, resolver.Address{
			Addr:       server.Addr,
			Attributes: attributes.New("weight", server.Weight),
		})
	}
	if len(r.srvAddrList) == 0 {
		logs.Error("no services found.")
		return nil
	}
	// 3. 设置 gRPC 客户端列表
	if err := r.clientConn.UpdateState(resolver.State{
		Addresses: r.srvAddrList,
	}); err != nil {
		logs.Fatal("grpc client update etcd state failed,err:%v name:%s", err, r.key)
	}
	return nil
}

// watch 对 etcd 的节点信息进行监听
func (r *Resolver) watch() {
	// 1. 定时一分钟同步一次数据
	ticker := time.NewTicker(time.Minute)
	// 2. 监听 etcd 事件
	r.watchChan = r.etcdClient.Watch(context.Background(), r.key, clientv3.WithPrefix())
	for {
		select {
		case <-ticker.C:
			if err := r.sync(); err != nil {
				logs.Error("watch sync failed,err:%v", err)
			}
		case response, ok := <-r.watchChan:
			if ok {
				r.update(response.Events)
			}
		case <-r.closeChan:
			if err := r.etcdClient.Close(); err != nil {
				logs.Error("close etcd client failed,err:%v", err)
			}
		}
	}

}

// update 监听 etcd 事件并修改本地服务列表
func (r *Resolver) update(events []*clientv3.Event) {
	for _, event := range events {
		switch event.Type {
		case clientv3.EventTypePut:
			server, err := ParseValue(event.Kv.Value)
			if err != nil {
				logs.Error("grpc client parse etcd value failed,err:%v value:%s", err, event.Kv.Value)
				continue
			}
			addr := resolver.Address{
				Addr:       server.Addr,
				Attributes: attributes.New("weight", server.Weight),
			}
			if !Exist(r.srvAddrList, addr) {
				// 更新列表
				r.srvAddrList = append(r.srvAddrList)
				err = r.clientConn.UpdateState(resolver.State{Addresses: r.srvAddrList})
				if err != nil {
					logs.Error("grpc client update parse etcd value faild,err:%v name:%s", err, string(event.Kv.Value))
				}
			}
		case clientv3.EventTypeDelete: // 删除事件
			// 解析为 server
			server, err := ParseKey(string(event.Kv.Key))
			if err != nil {
				logs.Error("grpc client update parse etcd value faild,err:%v name:%s", err, string(event.Kv.Key))
				continue
			}
			// 在列表中找到事件的 addr 并移除
			addr := resolver.Address{Addr: server.Addr}
			for i := range r.srvAddrList {
				if r.srvAddrList[i].Addr == addr.Addr {
					r.srvAddrList[i] = r.srvAddrList[len(r.srvAddrList)-1]
				}
			}
			// 更新列表
			if err := r.clientConn.UpdateState(resolver.State{
				Addresses: r.srvAddrList,
			}); err != nil {
				logs.Fatal("grpc client update etcd state failed,err:%v name:%s", err, r.key)
			}
		}
	}
}

func (r *Resolver) Close() {
	r.closeChan <- struct{}{}
}

func Exist(list []resolver.Address, addr resolver.Address) bool {
	for i := range list {
		if list[i].Addr == addr.Addr {
			return true
		}
	}
	return false
}

func NewResolver(config config.EtcdConf) *Resolver {
	return &Resolver{
		conf: config,
	}
}
